/*
 *
 *  *  Copyright 2010-2016 OrientDB LTD (http://orientdb.com)
 *  *
 *  *  Licensed under the Apache License, Version 2.0 (the "License");
 *  *  you may not use this file except in compliance with the License.
 *  *  You may obtain a copy of the License at
 *  *
 *  *       http://www.apache.org/licenses/LICENSE-2.0
 *  *
 *  *  Unless required by applicable law or agreed to in writing, software
 *  *  distributed under the License is distributed on an "AS IS" BASIS,
 *  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  *  See the License for the specific language governing permissions and
 *  *  limitations under the License.
 *  *
 *  * For more information: http://orientdb.com
 *
 */

package com.orientechnologies.orient.core.tx;

import com.orientechnologies.common.exception.OException;
import com.orientechnologies.common.log.OLogManager;
import com.orientechnologies.common.log.OLogger;
import com.orientechnologies.orient.core.db.ODatabaseDocumentInternal;
import com.orientechnologies.orient.core.db.document.RecordReader;
import com.orientechnologies.orient.core.db.record.OIdentifiable;
import com.orientechnologies.orient.core.db.record.ORecordOperation;
import com.orientechnologies.orient.core.exception.ODatabaseException;
import com.orientechnologies.orient.core.exception.ORecordNotFoundException;
import com.orientechnologies.orient.core.exception.OStorageException;
import com.orientechnologies.orient.core.exception.OTransactionException;
import com.orientechnologies.orient.core.hook.ORecordHook.TYPE;
import com.orientechnologies.orient.core.id.ORID;
import com.orientechnologies.orient.core.id.ORecordId;
import com.orientechnologies.orient.core.index.OClassIndexManager;
import com.orientechnologies.orient.core.index.OCompositeKey;
import com.orientechnologies.orient.core.index.OIndex;
import com.orientechnologies.orient.core.index.OIndexDefinition;
import com.orientechnologies.orient.core.index.OIndexManagerAbstract;
import com.orientechnologies.orient.core.metadata.schema.OClass;
import com.orientechnologies.orient.core.metadata.schema.OImmutableClass;
import com.orientechnologies.orient.core.metadata.schema.OType;
import com.orientechnologies.orient.core.metadata.sequence.OSessionSequenceLibrary;
import com.orientechnologies.orient.core.query.live.OLiveQueryHook;
import com.orientechnologies.orient.core.query.live.OLiveQueryHookV2;
import com.orientechnologies.orient.core.record.ORecord;
import com.orientechnologies.orient.core.record.ORecordInternal;
import com.orientechnologies.orient.core.record.impl.ODirtyManager;
import com.orientechnologies.orient.core.record.impl.ODocument;
import com.orientechnologies.orient.core.record.impl.ODocumentInternal;
import com.orientechnologies.orient.core.schedule.OScheduledEvent;
import com.orientechnologies.orient.core.serialization.serializer.record.ORecordSerializer;
import com.orientechnologies.orient.core.storage.OStorageRecordOperation;
import com.orientechnologies.orient.core.storage.OStorageTransactionIndexChanges;
import com.orientechnologies.orient.core.tx.OTransactionIndexChanges.OPERATION;
import com.orientechnologies.orient.core.tx.OTransactionIndexChangesPerKey.OTransactionIndexEntry;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;

public class OTransactionOptimistic extends OTransactionAbstract implements OTransactionInternal {
  private static final OLogger logger = OLogManager.instance().logger(OTransactionOptimistic.class);
  protected Map<ORID, ORID> updatedRids = new HashMap<>();
  protected Map<ORID, ORecordOperation> allEntries = new LinkedHashMap<>();
  protected Map<String, OTransactionIndexChanges> indexEntries = new LinkedHashMap<>();
  protected Map<ORID, List<OTransactionRecordIndexOperation>> recordIndexOperations =
      new HashMap<>();
  protected int id;
  protected int newObjectCounter = -2;
  protected Map<String, Object> userData = new HashMap<>();
  private Map<ORID, LockedRecordMetadata> noTxLocks;
  private Optional<OTxMetadataHolder> metadata = Optional.empty();

  /**
   * token This set is used to track which documents are changed during tx, if documents are changed
   * but not saved all changes are made during tx will be undone.
   */
  protected final Set<ODocument> changedDocuments = new HashSet<ODocument>();

  private Optional<List<byte[]>> serializedOperations = Optional.empty();

  private static final AtomicInteger txSerial = new AtomicInteger();
  protected boolean changed = true;
  private boolean alreadyCleared = false;
  private int txStartCounter;
  private boolean sentToServer = false;

  public OTransactionOptimistic(final ODatabaseDocumentInternal iDatabase) {
    super(iDatabase);
    this.id = txSerial.incrementAndGet();
  }

  public void begin() {
    if (txStartCounter < 0) {
      throw new OTransactionException("Invalid value of TX counter: " + txStartCounter);
    }
    if (txStartCounter == 0) {
      status = TXSTATUS.BEGUN;
    }
    txStartCounter++;

    if (txStartCounter > 1) {
      logger.debug(
          "Transaction with ID %d" + " was already started %d" + " times, and will be reused.",
          getId(), txStartCounter);
    }
  }

  public void commit() {
    commit(false);
  }

  /**
   * The transaction is reentrant. If {@code begin()} has been called several times, the actual
   * commit happens only after the same amount of {@code commit()} calls
   *
   * @param force commit transaction even
   */
  @Override
  public void commit(final boolean force) {
    checkTransactionValid();
    if (txStartCounter < 0) {
      throw new OStorageException("Invalid value of tx counter: " + txStartCounter);
    }
    if (force) {
      txStartCounter = 0;
    } else {
      txStartCounter--;
    }

    if (txStartCounter == 0) {
      doCommit();
    } else if (txStartCounter > 0) {
      logger.debug("Nested transaction was closed but transaction itself was not committed.");
    } else {
      throw new OTransactionException("Transaction was committed more times than it was started.");
    }
  }

  public void commitPreallocate() {
    checkTransactionValid();
    if (txStartCounter < 0) {
      throw new OStorageException("Invalid value of tx counter: " + txStartCounter);
    }
    txStartCounter--;

    if (txStartCounter == 0) {
      if (status == TXSTATUS.ROLLED_BACK || status == TXSTATUS.ROLLBACKING) {
        throw new ORollbackException(
            "Given transaction was rolled back, and thus cannot be committed.");
      }
      status = TXSTATUS.COMMITTING;

      if (sentToServer || !allEntries.isEmpty() || !indexEntries.isEmpty()) {
        database.internalCommitPreallocate(this);
      }
      close();
      status = TXSTATUS.COMPLETED;
    } else if (txStartCounter > 0) {
      logger.debug("Nested transaction was closed but transaction itself was not committed.");
    } else {
      throw new OTransactionException("Transaction was committed more times than it was started.");
    }
  }

  @Override
  public int amountOfNestedTxs() {
    return txStartCounter;
  }

  public void rollback() {
    rollback(false, -1);
  }

  public void internalRollback() {
    status = TXSTATUS.ROLLBACKING;

    // REMOVE ALL THE DIRTY ENTRIES AND UNDO ANY DIRTY DOCUMENT IF POSSIBLE.
    for (final ORecordOperation v : allEntries.values()) {
      final ORecord rec = v.getRecord();
      rec.unload();
      database.getLocalCache().deleteRecord(rec.getIdentity());
    }
    close();
    status = TXSTATUS.ROLLED_BACK;
  }

  @Override
  public void rollback(boolean force, int commitLevelDiff) {
    if (txStartCounter < 0) {
      throw new OStorageException("Invalid value of TX counter");
    }
    checkTransactionValid();

    txStartCounter += commitLevelDiff;
    status = TXSTATUS.ROLLBACKING;

    if (!force && txStartCounter > 0) {
      logger.debug(
          "Nested transaction was closed but transaction itself was scheduled for rollback.");
      return;
    }

    if (txStartCounter < 0) {
      throw new OTransactionException(
          "Transaction was rolled back more times than it was started.");
    }

    if (database.isRemote()) {
      ((ODatabaseDocumentInternal) database).remoteRollback(OTransactionOptimistic.this);
    }
    internalRollback();
  }

  public ORecord loadRecord(
      final ORID rid,
      final ORecord iRecord,
      final String fetchPlan,
      final boolean ignoreCache,
      final boolean iUpdateCache) {
    checkTransactionValid();

    final ORecord txRecord = getRecord(rid);
    if (txRecord == OTransactionAbstract.DELETED_RECORD) {
      // DELETED IN TX
      return null;
    }

    if (txRecord != null) {
      if (iRecord != null && txRecord != iRecord) {
        logger.warn(
            "Found record in transaction with the same RID %s but different instance. Probably"
                + " the record has been loaded from another transaction and reused on the"
                + " current one: reload it from current transaction before to update or delete"
                + " it.",
            iRecord.getIdentity());
      }
      return txRecord;
    }

    if (rid.isTemporary()) {
      return null;
    }

    final ORecord record =
        database.executeReadRecord(
            (ORecordId) rid,
            iRecord,
            -1,
            fetchPlan,
            ignoreCache,
            iUpdateCache,
            database::directRead);

    if (record != null && isolationLevel == ISOLATION_LEVEL.REPEATABLE_READ) {
      // KEEP THE RECORD IN TX TO ASSURE REPEATABLE READS
      addRecord(record, ORecordOperation.LOADED, null);
    }
    return record;
  }

  @Override
  public ORecord loadRecordIfVersionIsNotLatest(
      ORID rid, final int recordVersion, String fetchPlan, boolean ignoreCache)
      throws ORecordNotFoundException {
    checkTransactionValid();

    final ORecord txRecord = getRecord(rid);
    if (txRecord == OTransactionAbstract.DELETED_RECORD) {
      // DELETED IN TX
      throw new ORecordNotFoundException(rid);
    }

    if (txRecord != null) {
      if (txRecord.getVersion() > recordVersion) {
        return txRecord;
      } else {
        return null;
      }
    }
    if (rid.isTemporary()) {
      throw new ORecordNotFoundException(rid);
    }

    final ORecord record =
        database.executeReadRecord(
            (ORecordId) rid,
            null,
            recordVersion,
            fetchPlan,
            ignoreCache,
            !ignoreCache,
            database::readIfVersionIsNotLatest);

    if (record != null && isolationLevel == ISOLATION_LEVEL.REPEATABLE_READ) {
      // KEEP THE RECORD IN TX TO ASSURE REPEATABLE READS
      addRecord(record, ORecordOperation.LOADED, null);
    }
    return record;
  }

  @Override
  public ORecord reloadRecord(ORID rid, ORecord iRecord, String fetchPlan, boolean ignoreCache) {
    return reloadRecord(rid, iRecord, fetchPlan, ignoreCache, true);
  }

  @Override
  public ORecord reloadRecord(
      ORID rid, ORecord passedRecord, String fetchPlan, boolean ignoreCache, boolean force) {
    checkTransactionValid();

    final ORecord txRecord = getRecord(rid);
    if (txRecord == OTransactionAbstract.DELETED_RECORD) {
      // DELETED IN TX
      return null;
    }

    if (txRecord != null) {
      if (passedRecord != null && txRecord != passedRecord) {
        logger.warn(
            "Found record in transaction with the same RID %s but different instance. Probably"
                + " the record has been loaded from another transaction and reused on the"
                + " current one: reload it from current transaction before to update or delete"
                + " it",
            passedRecord.getIdentity());
      }
      return txRecord;
    }

    if (rid.isTemporary()) {
      return null;
    }

    final ORecord record;
    try {
      final RecordReader recordReader;
      if (force) {
        recordReader = database::directRead;
      } else {
        recordReader = database::readIfVersionIsNotLatest;
      }

      final ORecord loadedRecord =
          database.executeReadRecord(
              (ORecordId) rid,
              passedRecord,
              -1,
              fetchPlan,
              ignoreCache,
              !ignoreCache,
              recordReader);

      if (force) {
        record = loadedRecord;
      } else {
        if (loadedRecord == null) {
          record = passedRecord;
        } else {
          record = loadedRecord;
        }
      }
    } catch (final ORecordNotFoundException ignore) {
      return null;
    }

    if (record != null && isolationLevel == ISOLATION_LEVEL.REPEATABLE_READ) {
      // KEEP THE RECORD IN TX TO ASSURE REPEATABLE READS
      addRecord(record, ORecordOperation.LOADED, null);
    }
    return record;
  }

  @Override
  public ORecord loadRecord(ORID rid, ORecord record, String fetchPlan, boolean ignoreCache) {
    return loadRecord(rid, record, fetchPlan, ignoreCache, true);
  }

  public void deleteRecord(final ORecord iRecord) {
    if (!iRecord.getIdentity().isValid()) return;
    Set<ORecord> records = ORecordInternal.getDirtyManager(iRecord).getUpdateRecords();
    if (records != null) {
      for (final ORecord rec : records) {
        saveRecord(rec, null, false);
      }
    }

    final Set<ORecord> newRecords = ORecordInternal.getDirtyManager(iRecord).getNewRecords();
    if (newRecords != null) {
      for (final ORecord rec : newRecords) {
        saveRecord(rec, null, false);
      }
    }
    addRecord(iRecord, ORecordOperation.DELETED, null);
  }

  public ORecord saveRecord(
      final ORecord iRecord, final String iClusterName, final boolean iForceCreate) {
    if (iRecord == null) {
      return null;
    }
    boolean originalSaved = false;
    final ODirtyManager dirtyManager = ORecordInternal.getDirtyManager(iRecord);
    do {
      final Set<ORecord> newRecord = dirtyManager.getNewRecords();
      final Set<ORecord> updatedRecord = dirtyManager.getUpdateRecords();
      dirtyManager.clear();
      if (newRecord != null) {
        for (ORecord rec : newRecord) {
          if (rec instanceof ODocument)
            ODocumentInternal.convertAllMultiValuesToTrackedVersions((ODocument) rec);
          if (rec == iRecord) {
            addRecord(rec, ORecordOperation.CREATED, iClusterName);
            originalSaved = true;
          } else addRecord(rec, ORecordOperation.CREATED, database.getClusterName(rec));
        }
      }
      if (updatedRecord != null) {
        for (ORecord rec : updatedRecord) {
          if (rec instanceof ODocument)
            ODocumentInternal.convertAllMultiValuesToTrackedVersions((ODocument) rec);
          if (rec == iRecord) {
            final byte operation;
            if (iForceCreate) {
              operation = ORecordOperation.CREATED;
            } else {
              operation =
                  iRecord.getIdentity().isValid()
                      ? ORecordOperation.UPDATED
                      : ORecordOperation.CREATED;
            }
            addRecord(rec, operation, iClusterName);
            originalSaved = true;
          } else addRecord(rec, ORecordOperation.UPDATED, database.getClusterName(rec));
        }
      }
    } while (dirtyManager.getNewRecords() != null || dirtyManager.getUpdateRecords() != null);

    if (!originalSaved && iRecord.isDirty()) {
      final byte operation;
      if (iForceCreate) {
        operation = ORecordOperation.CREATED;
      } else {
        operation =
            iRecord.getIdentity().isValid() ? ORecordOperation.UPDATED : ORecordOperation.CREATED;
      }
      addRecord(iRecord, operation, iClusterName);
    }
    return iRecord;
  }

  @Override
  public String toString() {
    return "OTransactionOptimistic [id="
        + id
        + ", status="
        + status
        + ", recEntries="
        + allEntries.size()
        + ", idxEntries="
        + indexEntries.size()
        + ']';
  }

  public void setStatus(final TXSTATUS iStatus) {
    status = iStatus;
  }

  public ORecordOperation addRecord(ORecord iRecord, byte iStatus, String iClusterName) {
    changed = true;
    checkTransactionValid();

    if (iClusterName == null) {
      iClusterName = database.getClusterNameById(iRecord.getIdentity().getClusterId());
    }
    if (iStatus != ORecordOperation.LOADED) {
      changedDocuments.remove(iRecord);
    }

    try {
      final ORecordId rid = (ORecordId) iRecord.getIdentity();
      ORecordOperation txEntry = getRecordEntry(rid);
      if (iStatus == ORecordOperation.CREATED && txEntry != null) {
        iStatus = ORecordOperation.UPDATED;
      }
      switch (iStatus) {
        case ORecordOperation.CREATED:
          {
            OIdentifiable res = database.beforeCreateOperations(iRecord, iClusterName);
            if (res != null) {
              iRecord = (ORecord) res;
            }
          }
          break;
        case ORecordOperation.LOADED:
          /** Read hooks already invoked in {@link ODatabaseDocumentTx#executeReadRecord} */
          break;
        case ORecordOperation.UPDATED:
          {
            OIdentifiable res = database.beforeUpdateOperations(iRecord, iClusterName);
            if (res != null) {
              iRecord = (ORecord) res;
            }
          }
          break;
        case ORecordOperation.DELETED:
          database.beforeDeleteOperations(iRecord, iClusterName);
          break;
      }

      try {
        if (!rid.isValid()) {
          ORecordInternal.onBeforeIdentityChanged(iRecord);
          database.assignAndCheckCluster(iRecord, iClusterName);

          rid.setClusterPosition(newObjectCounter--);

          ORecordInternal.onAfterIdentityChanged(iRecord);
        }
        if (txEntry == null) {
          if (!(rid.isTemporary() && iStatus != ORecordOperation.CREATED)) {
            // NEW ENTRY: JUST REGISTER IT
            txEntry = new ORecordOperation(iRecord, iStatus);
            allEntries.put(rid.copy(), txEntry);
          }
        } else {
          // UPDATE PREVIOUS STATUS
          txEntry.record = iRecord;

          switch (txEntry.getType()) {
            case ORecordOperation.LOADED:
              switch (iStatus) {
                case ORecordOperation.UPDATED:
                  txEntry.setType(ORecordOperation.UPDATED);
                  break;
                case ORecordOperation.DELETED:
                  txEntry.setType(ORecordOperation.DELETED);
                  break;
              }
              break;
            case ORecordOperation.UPDATED:
              switch (iStatus) {
                case ORecordOperation.DELETED:
                  txEntry.setType(ORecordOperation.DELETED);
                  break;
              }
              break;
            case ORecordOperation.DELETED:
              break;
            case ORecordOperation.CREATED:
              switch (iStatus) {
                case ORecordOperation.DELETED:
                  allEntries.remove(rid);
                  // txEntry.type = ORecordOperation.DELETED;
                  break;
              }
              break;
          }
        }

        switch (iStatus) {
          case ORecordOperation.CREATED:
            database.afterCreateOperations(iRecord);
            break;
          case ORecordOperation.LOADED:
            /** Read hooks already invoked in {@link ODatabaseDocumentTx#executeReadRecord} . */
            break;
          case ORecordOperation.UPDATED:
            database.afterUpdateOperations(iRecord);
            break;
          case ORecordOperation.DELETED:
            database.afterDeleteOperations(iRecord);
            break;
        }

        // RESET TRACKING
        if (iRecord instanceof ODocument && ((ODocument) iRecord).isTrackingChanges()) {
          ODocumentInternal.clearTrackData(((ODocument) iRecord));
        }
        return txEntry;
      } catch (final Exception e) {
        switch (iStatus) {
          case ORecordOperation.CREATED:
            database.callbackHooks(TYPE.CREATE_FAILED, iRecord);
            break;
          case ORecordOperation.UPDATED:
            database.callbackHooks(TYPE.UPDATE_FAILED, iRecord);
            break;
          case ORecordOperation.DELETED:
            database.callbackHooks(TYPE.DELETE_FAILED, iRecord);
            break;
        }
        throw OException.wrapException(
            new ODatabaseException("Error on saving record " + iRecord.getIdentity()), e);
      }
    } finally {
      switch (iStatus) {
        case ORecordOperation.CREATED:
          database.callbackHooks(TYPE.FINALIZE_CREATION, iRecord);
          break;
        case ORecordOperation.UPDATED:
          database.callbackHooks(TYPE.FINALIZE_UPDATE, iRecord);
          break;
        case ORecordOperation.DELETED:
          database.callbackHooks(TYPE.FINALIZE_DELETION, iRecord);
          break;
      }
    }
  }

  private void doCommit() {
    if (status == TXSTATUS.ROLLED_BACK || status == TXSTATUS.ROLLBACKING) {
      throw new ORollbackException(
          "Given transaction was rolled back, and thus cannot be committed.");
    }
    status = TXSTATUS.COMMITTING;

    if (sentToServer || !allEntries.isEmpty() || !indexEntries.isEmpty()) {
      database.internalCommit(this);
    }
    close();
    status = TXSTATUS.COMPLETED;
  }

  public void resetChangesTracking() {
    alreadyCleared = true;
    changed = false;
  }

  public boolean isChanged() {
    return changed;
  }

  public boolean isAlreadyCleared() {
    return alreadyCleared;
  }

  public Set<ORID> getLockedRecords() {
    if (getNoTxLocks() != null) {
      final Set<ORID> rids = new HashSet<ORID>(getNoTxLocks().keySet());
      rids.addAll(locks.keySet());
      return rids;
    } else {
      return locks.keySet();
    }
  }

  public void setSentToServer(boolean sentToServer) {
    this.sentToServer = sentToServer;
  }

  public boolean getSentToServer() {
    return sentToServer;
  }

  public void fill(final Iterator<ORecordOperation> operations) {
    while (operations.hasNext()) {
      ORecordOperation change = operations.next();
      allEntries.put(change.getRID(), change);
      resolveTracking(change);
    }
  }

  protected void resolveTracking(final ORecordOperation change) {
    if (!(change.getRecord() instanceof ODocument)) {
      return;
    }
    final ODocument rec = (ODocument) change.getRecord();
    switch (change.getType()) {
      case ORecordOperation.CREATED:
        {
          final ODocument doc = (ODocument) change.getRecord();
          OLiveQueryHook.addOp(doc, ORecordOperation.CREATED, database);
          OLiveQueryHookV2.addOp(doc, ORecordOperation.CREATED, database);
          final OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(doc);
          if (clazz != null) {
            OClassIndexManager.processIndexOnCreate(database, rec);
            if (clazz.isFunction()) {
              database.getSharedContext().getFunctionLibrary().createdFunction(doc);
            }
            if (clazz.isSequence()) {
              ((OSessionSequenceLibrary) database.getMetadata().getSequenceLibrary())
                  .getDelegate()
                  .onSequenceCreated(database, doc);
            }
            if (clazz.isScheduler()) {
              database.getMetadata().getScheduler().scheduleEvent(new OScheduledEvent(doc));
            }
          }
        }
        break;
      case ORecordOperation.UPDATED:
        {
          final OIdentifiable updateRecord = change.getRecord();
          if (updateRecord instanceof ODocument) {
            final ODocument updateDoc = (ODocument) updateRecord;
            OLiveQueryHook.addOp(updateDoc, ORecordOperation.UPDATED, database);
            OLiveQueryHookV2.addOp(updateDoc, ORecordOperation.UPDATED, database);
            final OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(updateDoc);
            if (clazz != null) {
              OClassIndexManager.processIndexOnUpdate(database, updateDoc);
              if (clazz.isFunction()) {
                database.getSharedContext().getFunctionLibrary().updatedFunction(updateDoc);
              }
              if (clazz.isSequence()) {
                ((OSessionSequenceLibrary) database.getMetadata().getSequenceLibrary())
                    .getDelegate()
                    .onSequenceUpdated(database, updateDoc);
              }
            }
          }
        }
        break;
      case ORecordOperation.DELETED:
        {
          final ODocument doc = (ODocument) change.getRecord();
          final OImmutableClass clazz = ODocumentInternal.getImmutableSchemaClass(doc);
          if (clazz != null) {
            OClassIndexManager.processIndexOnDelete(database, rec);
            if (clazz.isFunction()) {
              database.getSharedContext().getFunctionLibrary().droppedFunction(doc);
              database
                  .getSharedContext()
                  .getOrientDB()
                  .getScriptManager()
                  .close(database.getName());
            }
            if (clazz.isSequence()) {
              ((OSessionSequenceLibrary) database.getMetadata().getSequenceLibrary())
                  .getDelegate()
                  .onSequenceDropped(database, doc);
            }
            if (clazz.isScheduler()) {
              final String eventName = doc.field(OScheduledEvent.PROP_NAME);
              database.getSharedContext().getScheduler().removeEventInternal(eventName);
            }
          }
          OLiveQueryHook.addOp(doc, ORecordOperation.DELETED, database);
          OLiveQueryHookV2.addOp(doc, ORecordOperation.DELETED, database);
        }
        break;
      case ORecordOperation.LOADED:
      default:
        break;
    }
  }

  protected int getTxStartCounter() {
    return txStartCounter;
  }

  @Override
  public void addChangedDocument(ODocument document) {
    if (getRecord(document.getIdentity()) == null) {
      changedDocuments.add(document);
    }
  }

  public int getId() {
    return id;
  }

  @Override
  public int getEntryCount() {
    return allEntries.size();
  }

  public Collection<ORecordOperation> getRecordOperations() {
    return allEntries.values();
  }

  public ORecordOperation getRecordEntry(ORID ridPar) {
    ORID rid = ridPar;
    ORecordOperation entry;
    do {
      entry = allEntries.get(rid);
      if (entry == null) {
        rid = updatedRids.get(rid);
      }
    } while (entry == null && rid != null && !rid.equals(ridPar));
    return entry;
  }

  public ORecord getRecord(final ORID rid) {
    final ORecordOperation e = getRecordEntry(rid);
    if (e != null)
      if (e.getType() == ORecordOperation.DELETED) return OTransactionAbstract.DELETED_RECORD;
      else return e.getRecord();
    return null;
  }

  /** Called by class iterator. */
  public List<ORecordOperation> getNewRecordEntriesByClass(
      final OClass iClass, final boolean iPolymorphic) {
    final List<ORecordOperation> result = new ArrayList<ORecordOperation>();

    if (iClass == null)
      // RETURN ALL THE RECORDS
      for (ORecordOperation entry : allEntries.values()) {
        if (entry.getType() == ORecordOperation.CREATED) result.add(entry);
      }
    else {
      // FILTER RECORDS BY CLASSNAME
      for (ORecordOperation entry : allEntries.values()) {
        if (entry.getType() == ORecordOperation.CREATED)
          if (entry.getRecord() != null && entry.getRecord() instanceof ODocument) {
            if (iPolymorphic) {
              if (iClass.isSuperClassOf(
                  ODocumentInternal.getImmutableSchemaClass(((ODocument) entry.getRecord()))))
                result.add(entry);
            } else if (iClass.getName().equals(((ODocument) entry.getRecord()).getClassName()))
              result.add(entry);
          }
      }
    }

    return result;
  }

  /** Called by cluster iterator. */
  public List<ORecordOperation> getNewRecordEntriesByClusterIds(final int[] iIds) {
    final List<ORecordOperation> result = new ArrayList<ORecordOperation>();

    if (iIds == null)
      // RETURN ALL THE RECORDS
      for (ORecordOperation entry : allEntries.values()) {
        if (entry.getType() == ORecordOperation.CREATED) result.add(entry);
      }
    else
      // FILTER RECORDS BY ID
      for (ORecordOperation entry : allEntries.values()) {
        for (int id : iIds) {
          if (entry.getRecord() != null
              && entry.getRecord().getIdentity().getClusterId() == id
              && entry.getType() == ORecordOperation.CREATED) {
            result.add(entry);
            break;
          }
        }
      }

    return result;
  }

  public List<String> getInvolvedIndexes() {
    List<String> list = null;
    for (String indexName : indexEntries.keySet()) {
      if (list == null) list = new ArrayList<String>();
      list.add(indexName);
    }
    return list;
  }

  public ODocument getIndexChanges() {

    final ODocument result = new ODocument().setAllowChainedAccess(false).setTrackingChanges(false);

    for (Entry<String, OTransactionIndexChanges> indexEntry : indexEntries.entrySet()) {
      final ODocument indexDoc = new ODocument().setTrackingChanges(false);
      ODocumentInternal.addOwner(indexDoc, result);

      result.field(indexEntry.getKey(), indexDoc, OType.EMBEDDED);

      if (indexEntry.getValue().cleared) indexDoc.field("clear", Boolean.TRUE);

      final List<ODocument> entries = new ArrayList<ODocument>();
      indexDoc.field("entries", entries, OType.EMBEDDEDLIST);

      // STORE INDEX ENTRIES
      for (OTransactionIndexChangesPerKey entry : indexEntry.getValue().changesPerKey.values()) {
        if (!entry.clientTrackOnly) entries.add(serializeIndexChangeEntry(entry, indexDoc));
      }

      indexDoc.field(
          "nullEntries", serializeIndexChangeEntry(indexEntry.getValue().nullKeyChanges, indexDoc));
    }

    indexEntries.clear();

    return result;
  }

  public Map<String, OTransactionIndexChanges> getIndexOperations() {
    return indexEntries;
  }

  /**
   * Bufferizes index changes to be flushed at commit time.
   *
   * @return
   */
  public OTransactionIndexChanges getIndexChanges(final String iIndexName) {
    return indexEntries.get(iIndexName);
  }

  public OTransactionIndexChanges getIndexChangesInternal(final String indexName) {
    if (getDatabase().isRemote()) return null;
    return getIndexChanges(indexName);
  }

  public void addIndexEntry(
      final OIndex delegate,
      final String iIndexName,
      final OTransactionIndexChanges.OPERATION iOperation,
      final Object key,
      final OIdentifiable iValue) {
    addIndexEntry(delegate, iIndexName, iOperation, key, iValue, false);
  }

  /** Bufferizes index changes to be flushed at commit time. */
  public void addIndexEntry(
      final OIndex delegate,
      final String iIndexName,
      final OTransactionIndexChanges.OPERATION iOperation,
      final Object key,
      final OIdentifiable iValue,
      boolean clientTrackOnly) {
    this.changed = true;
    OTransactionIndexChanges indexEntry = indexEntries.get(iIndexName);
    if (indexEntry == null) {
      indexEntry = new OTransactionIndexChanges();
      indexEntries.put(iIndexName, indexEntry);
    }

    if (iOperation == OPERATION.CLEAR) indexEntry.setCleared();
    else {
      OTransactionIndexChangesPerKey changes = indexEntry.getChangesPerKey(key);
      changes.clientTrackOnly = clientTrackOnly;
      changes.add(iValue, iOperation);

      if (iValue == null) return;

      List<OTransactionRecordIndexOperation> transactionIndexOperations =
          recordIndexOperations.get(iValue.getIdentity());

      if (transactionIndexOperations == null) {
        transactionIndexOperations = new ArrayList<OTransactionRecordIndexOperation>();
        recordIndexOperations.put(iValue.getIdentity().copy(), transactionIndexOperations);
      }

      transactionIndexOperations.add(
          new OTransactionRecordIndexOperation(iIndexName, key, iOperation));
    }
  }

  public void updateIdentityAfterCommit(final ORID oldRid, final ORID newRid) {
    if (oldRid.equals(newRid))
      // NO CHANGE, IGNORE IT
      return;

    // XXX: Identity update may mutate the index keys, so we have to identify and reinsert
    // potentially affected index keys to keep
    // the OTransactionIndexChanges.changesPerKey in a consistent state.

    final List<KeyChangesUpdateRecord> keyRecordsToReinsert =
        new ArrayList<KeyChangesUpdateRecord>();
    final ODatabaseDocumentInternal database = getDatabase();
    final OIndexManagerAbstract indexManager = database.getMetadata().getIndexManagerInternal();
    for (Entry<String, OTransactionIndexChanges> entry : indexEntries.entrySet()) {
      final OIndex index = indexManager.getIndex(database, entry.getKey());
      if (index == null)
        throw new OTransactionException(
            "Cannot find index '" + entry.getValue() + "' while committing transaction");

      final Dependency[] fieldRidDependencies = getIndexFieldRidDependencies(index);
      if (!isIndexMayDependOnRids(fieldRidDependencies)) continue;

      final OTransactionIndexChanges indexChanges = entry.getValue();
      for (final Iterator<OTransactionIndexChangesPerKey> iterator =
              indexChanges.changesPerKey.values().iterator();
          iterator.hasNext(); ) {
        final OTransactionIndexChangesPerKey keyChanges = iterator.next();
        if (isIndexKeyMayDependOnRid(keyChanges.key, oldRid, fieldRidDependencies)) {
          keyRecordsToReinsert.add(new KeyChangesUpdateRecord(keyChanges, indexChanges));
          iterator.remove();
        }
      }
    }

    // Update the identity.

    final ORecordOperation rec = getRecordEntry(oldRid);
    if (rec != null) {
      updatedRids.put(newRid.copy(), oldRid.copy());

      if (!rec.getRecord().getIdentity().equals(newRid)) {
        ORecordInternal.onBeforeIdentityChanged(rec.getRecord());

        final ORecordId recordId = (ORecordId) rec.getRecord().getIdentity();
        if (recordId == null) {
          ORecordInternal.setIdentity(rec.getRecord(), new ORecordId(newRid));
        } else {
          recordId.setClusterPosition(newRid.getClusterPosition());
          recordId.setClusterId(newRid.getClusterId());
        }

        ORecordInternal.onAfterIdentityChanged(rec.getRecord());
      }
    }

    // Reinsert the potentially affected index keys.

    for (KeyChangesUpdateRecord record : keyRecordsToReinsert)
      record.indexChanges.changesPerKey.put(record.keyChanges.key, record.keyChanges);

    // Update the indexes.

    ORecordOperation val = getRecordEntry(oldRid);
    final List<OTransactionRecordIndexOperation> transactionIndexOperations =
        recordIndexOperations.get(val != null ? val.getRID() : null);
    if (transactionIndexOperations != null) {
      for (final OTransactionRecordIndexOperation indexOperation : transactionIndexOperations) {
        OTransactionIndexChanges indexEntryChanges = indexEntries.get(indexOperation.index);
        if (indexEntryChanges == null) continue;
        final OTransactionIndexChangesPerKey keyChanges;
        if (indexOperation.key == null) {
          keyChanges = indexEntryChanges.nullKeyChanges;
        } else {
          keyChanges = indexEntryChanges.changesPerKey.get(indexOperation.key);
        }
        if (keyChanges != null) updateChangesIdentity(oldRid, newRid, keyChanges);
      }
    }
  }

  protected void checkTransactionValid() {
    if (status == TXSTATUS.INVALID) {
      throw new OTransactionException(
          "Invalid state of the transaction. The transaction must be begun.");
    }
  }

  protected ODocument serializeIndexChangeEntry(
      OTransactionIndexChangesPerKey entry, final ODocument indexDoc) {
    // SERIALIZE KEY

    ODocument keyContainer = new ODocument();
    keyContainer.setTrackingChanges(false);

    if (entry.key != null) {
      if (entry.key instanceof OCompositeKey) {
        final List<Object> keys = ((OCompositeKey) entry.key).getKeys();

        keyContainer.field("key", keys, OType.EMBEDDEDLIST);
        keyContainer.field("binary", false);
      } else {
        keyContainer.field("key", entry.key);
        keyContainer.field("binary", false);
      }

    } else keyContainer = null;

    final List<ODocument> operations = new ArrayList<ODocument>();

    // SERIALIZE VALUES
    if (!entry.isEmpty()) {
      for (OTransactionIndexEntry e : entry.getEntriesAsList()) {

        final ODocument changeDoc = new ODocument().setAllowChainedAccess(false);
        ODocumentInternal.addOwner(changeDoc, indexDoc);

        // SERIALIZE OPERATION
        changeDoc.field("o", e.getOperation().ordinal());

        if (e.getValue() instanceof ORecord && e.getValue().getIdentity().isNew()) {
          final ORecord saved = getRecord(e.getValue().getIdentity());
          if (saved != null) {
            e.setValue(saved);
          } else {
            database.save((ORecord) e.getValue());
          }
        }

        changeDoc.field("v", e.getValue() != null ? e.getValue().getIdentity() : null);

        operations.add(changeDoc);
      }
    }
    ODocument res = new ODocument();
    res.setTrackingChanges(false);
    ODocumentInternal.addOwner(res, indexDoc);
    return res.setAllowChainedAccess(false)
        .field("k", keyContainer, OType.EMBEDDED)
        .field("ops", operations, OType.EMBEDDEDLIST);
  }

  private void updateChangesIdentity(
      ORID oldRid, ORID newRid, OTransactionIndexChangesPerKey changesPerKey) {
    if (changesPerKey == null) return;

    for (final OTransactionIndexEntry indexEntry : changesPerKey.getEntriesAsList())
      if (indexEntry.getValue().getIdentity().equals(oldRid)) indexEntry.setValue(newRid);
  }

  @Override
  public void setCustomData(String iName, Object iValue) {
    userData.put(iName, iValue);
  }

  @Override
  public Object getCustomData(String iName) {
    return userData.get(iName);
  }

  private static Dependency[] getIndexFieldRidDependencies(OIndex index) {
    final OIndexDefinition definition = index.getDefinition();

    if (definition == null) // type for untyped index is still not resolved
    return null;

    final OType[] types = definition.getTypes();
    final Dependency[] dependencies = new Dependency[types.length];

    for (int i = 0; i < types.length; ++i) dependencies[i] = getTypeRidDependency(types[i]);

    return dependencies;
  }

  private static boolean isIndexMayDependOnRids(Dependency[] fieldDependencies) {
    if (fieldDependencies == null) return true;

    for (Dependency dependency : fieldDependencies)
      switch (dependency) {
        case Unknown:
        case Yes:
          return true;
        case No:
          break; // do nothing
      }

    return false;
  }

  private static boolean isIndexKeyMayDependOnRid(
      Object key, ORID rid, Dependency[] keyDependencies) {
    if (key instanceof OCompositeKey) {
      final List<Object> subKeys = ((OCompositeKey) key).getKeys();
      for (int i = 0; i < subKeys.size(); ++i)
        if (isIndexKeyMayDependOnRid(
            subKeys.get(i), rid, keyDependencies == null ? null : keyDependencies[i])) return true;
      return false;
    }

    return isIndexKeyMayDependOnRid(key, rid, keyDependencies == null ? null : keyDependencies[0]);
  }

  private static boolean isIndexKeyMayDependOnRid(Object key, ORID rid, Dependency dependency) {
    if (dependency == Dependency.No) return false;

    if (key instanceof OIdentifiable) return key.equals(rid);

    return dependency == Dependency.Unknown || dependency == null;
  }

  private static Dependency getTypeRidDependency(OType type) {
    switch (type) {
      case CUSTOM:
      case ANY:
        return Dependency.Unknown;

      case EMBEDDED:
      case LINK:
        return Dependency.Yes;

      case LINKLIST:
      case LINKSET:
      case LINKMAP:
      case LINKBAG:
      case EMBEDDEDLIST:
      case EMBEDDEDSET:
      case EMBEDDEDMAP:
        assert false; // under normal conditions, collection field type is already resolved to its
        // component type
        return Dependency.Unknown; // fallback to the safest variant, just in case

      default: // all other primitive types which doesn't depend on rids
        return Dependency.No;
    }
  }

  private enum Dependency {
    Unknown,
    Yes,
    No
  }

  private static class KeyChangesUpdateRecord {
    public final OTransactionIndexChangesPerKey keyChanges;
    public final OTransactionIndexChanges indexChanges;

    public KeyChangesUpdateRecord(
        OTransactionIndexChangesPerKey keyChanges, OTransactionIndexChanges indexChanges) {
      this.keyChanges = keyChanges;
      this.indexChanges = indexChanges;
    }
  }

  public Map<ORID, ORID> getUpdatedRids() {
    return updatedRids;
  }

  public int getNewObjectCounter() {
    return newObjectCounter;
  }

  public void setNoTxLocks(Map<ORID, LockedRecordMetadata> noTxLocks) {
    this.noTxLocks = noTxLocks;
  }

  public Map<ORID, LockedRecordMetadata> getNoTxLocks() {
    return noTxLocks;
  }

  @Override
  public Optional<byte[]> getMetadata() {
    return metadata.map((h) -> h.metadata());
  }

  @Override
  public void startedStorageTransaction() {
    if (metadata.isPresent()) {
      metadata.get().notifyMetadataRead();
    }
  }

  @Override
  public void setMetadataHolder(Optional<OTxMetadataHolder> metadata) {
    this.metadata = metadata;
  }

  @Override
  public void prepareSerializedOperations() throws IOException {
    List<byte[]> operations = new ArrayList<>();
    for (ORecordOperation value : allEntries.values()) {
      OTransactionDataChange change = new OTransactionDataChange(value);
      ByteArrayOutputStream out = new ByteArrayOutputStream();
      change.serialize(new DataOutputStream(out));
      operations.add(out.toByteArray());
    }
    this.serializedOperations = Optional.of(operations);
  }

  public Iterator<byte[]> getSerializedOperations() {
    if (serializedOperations.isPresent()) {
      return serializedOperations.get().iterator();
    } else {
      return Collections.emptyIterator();
    }
  }

  @Override
  public void resetAllocatedIds() {
    for (Map.Entry<ORID, ORecordOperation> op : allEntries.entrySet()) {
      if (op.getValue().getType() == ORecordOperation.CREATED) {
        ORID lastCreateId = op.getValue().getRID().copy();
        ORecordId oldNew =
            new ORecordId(lastCreateId.getClusterId(), op.getKey().getClusterPosition());
        updateIdentityAfterCommit(lastCreateId, oldNew);
        updatedRids.put(oldNew, op.getKey());
      }
    }
  }

  public void close() {
    super.close();
    for (final ORecordOperation recordOperation : getRecordOperations()) {
      final ORecord record = recordOperation.getRecord();
      if (record instanceof ODocument) {
        final ODocument document = (ODocument) record;
        if (document.isDirty()) {
          document.undo();
        }
        changedDocuments.remove(document);
      }
    }

    for (ODocument changedDocument : changedDocuments) {
      if (!changedDocument.isEmbedded()) {
        changedDocument.undo();
      }
    }
    changedDocuments.clear();
    updatedRids.clear();
    allEntries.clear();
    indexEntries.clear();
    recordIndexOperations.clear();
    newObjectCounter = -2;
    status = TXSTATUS.INVALID;

    database.setDefaultTransactionMode(getNoTxLocks());
    userData.clear();
  }

  @Override
  public void updateCache(boolean keepInCache) {
    OTransactionAbstract.updateCacheFromEntries(getDatabase(), getRecordOperations(), keepInCache);
  }

  @Override
  public SortedMap<String, OStorageTransactionIndexChanges> getChangesForIndex() {
    TreeMap<String, OStorageTransactionIndexChanges> map = new TreeMap<>();
    for (Map.Entry<String, OTransactionIndexChanges> change : this.indexEntries.entrySet()) {
      String index = change.getKey();
      OTransactionIndexChanges ops = change.getValue();
      ops.resolveAssociatedIndex(index, database);
      map.put(index, ops);
    }
    return map;
  }

  @Override
  public Collection<OStorageRecordOperation> getRecordChanges() {
    ORecordSerializer serializer = database.getSerializer();
    for (ORecordOperation op : allEntries.values()) {
      op.setSerializer(serializer);
    }

    return (Collection<OStorageRecordOperation>) (Collection) allEntries.values();
  }
}
